package com.zbj.storm.wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import java.util.concurrent.TimeUnit;

/**
 * WordCountTopology
 *
 * @author weigang
 * @create 2019-09-04
 **/
public class WordCountTopology {
    public static void main(String[] args) throws Exception {
        // 创建组件
        LineSpout spout = new LineSpout();
        SplitLineBolt splitLineBolt = new SplitLineBolt();
        WordCountBolt wordCountBolt = new WordCountBolt();
        ReportBolt reportBolt = new ReportBolt();

        // 创建拓扑构建者
        TopologyBuilder topologyBuilder = new TopologyBuilder();

        // 组织拓扑
        topologyBuilder.setSpout("LineSpout", spout, 1);
        topologyBuilder.setBolt("SplitLineBolt", splitLineBolt).shuffleGrouping("LineSpout");
        topologyBuilder.setBolt("WordCountBolt", wordCountBolt).fieldsGrouping("SplitLineBolt", new Fields("word"));
        topologyBuilder.setBolt("ReportBolt", reportBolt).globalGrouping("WordCountBolt");

        // 创建拓扑
        StormTopology topology = topologyBuilder.createTopology();

        // 创建配合对象
        Config config = new Config();

        if (args.length == 0) {
            config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
            config.setNumWorkers(2);

            // 创建本地集群
            LocalCluster localCluster = new LocalCluster();
            // 提交Topology给集群运行
            localCluster.submitTopology("word_count_topology", config, topology);

            TimeUnit.SECONDS.sleep(10);

            localCluster.killTopology("word_count_topology");
            localCluster.shutdown();
        } else {
            //集群模式
            try {
                StormSubmitter.submitTopology(args[0], config, topology);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}